/*
 * Copyright zhoujun 2021 Wicrenet, Inc. All rights reserved.
 */
package com.ruoyi.netty.handler;

import com.alibaba.fastjson.JSON;
import com.ruoyi.common.core.domain.model.LoginUser;
import com.ruoyi.framework.web.service.TokenService;
import com.ruoyi.netty.domain.DeviceChannelInfo;
import com.ruoyi.netty.domain.MsgAgreement;
import com.ruoyi.netty.utils.CacheService;
import com.ruoyi.netty.utils.CacheUtil;
import com.ruoyi.netty.utils.NetWorkUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.joda.time.LocalDateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 【websocket通讯处理】
 *
 * @author zhoujun
 * Created on 2021/11/25 11:23
 */
@Component
@ChannelHandler.Sharable
public class WebSocketServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> implements DisposableBean {

    private static final Logger       logger       = LoggerFactory.getLogger(WebSocketServerHandler.class);
    private static       ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    @Autowired
    private        CacheService cacheService;
    @Autowired
    private TokenService tokenService;


    /**
     * 【 读取客户端发送的数据 】
     *
     * @author zhoujun
     * @date 2021/11/26 15:31
     * @param ctx
     * @param msg
     * @return void
     **/
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        Channel localChannel = ctx.channel();
        SocketChannel socketChannel = (SocketChannel)localChannel;
        String text = msg.text();
        MsgAgreement msgAgreement = JSON.parseObject(text, MsgAgreement.class);
        String token = msgAgreement.getToken();

        LoginUser loginUser = tokenService.getLoginUserByToken(token);
        if (loginUser == null){
            logger.info("登录token失效:{}",token);
            return;
        }
        tokenService.verifyToken(loginUser);

        if(msgAgreement.getMessageType().equals(2)){
            List<DeviceChannelInfo> deviceChannelInfos = cacheService.getRedisUtil().popList();
            String devices = JSON.toJSONString(deviceChannelInfos);
            this.sendToClient(localChannel, devices);
            return;
        }

        if(msgAgreement.getMessageType().equals(1)){
            String toUserId = msgAgreement.getToUserId();
            //判断接收消息用户是否在本服务端
            Channel cacheChannel = CacheUtil.cacheChannel.get(toUserId);
            if (null != cacheChannel) {
                MsgAgreement msgAgreementNew = new MsgAgreement();
                msgAgreementNew.setMessageType(5);
                msgAgreementNew.setContent(msgAgreement.getContent());
                if (cacheChannel != localChannel) {
                    this.sendToClient(cacheChannel,JSON.toJSONString(msgAgreementNew));
                } else {
                    this.sendToClient(localChannel, "【自己】" + text);
                }
                return;
            }
            //如果为NULL则接收消息的用户不在本服务端，需要push消息给全局
            this.cacheService.push(msgAgreement);;
            return;
        }

        //心跳信息
        if(msgAgreement.getMessageType().equals(3)){
            logger.info("心跳信息:{}",msgAgreement.getContent());
            //更新设备信息
            DeviceChannelInfo deviceChannelInfo = new DeviceChannelInfo();
            deviceChannelInfo.setChannelId(socketChannel.id().toString());
            deviceChannelInfo.setIp(NetWorkUtils.getHost());
            deviceChannelInfo.setPort(socketChannel.localAddress().getPort());
            deviceChannelInfo.setLinkDate(new Date());
            deviceChannelInfo.setUserId(msgAgreement.getUserId());
            this.addClientCache(msgAgreement.getUserId(),socketChannel, deviceChannelInfo);
            this.sendToClient(localChannel, text);
        }

        //登录
        if(msgAgreement.getMessageType().equals(4)){
            logger.info("开始登录:{}",msgAgreement.getContent());
            //保存设备信息
            DeviceChannelInfo deviceChannelInfo = new DeviceChannelInfo();
            deviceChannelInfo.setChannelId(socketChannel.id().toString());
            deviceChannelInfo.setIp(NetWorkUtils.getHost());
            deviceChannelInfo.setPort(socketChannel.localAddress().getPort());
            deviceChannelInfo.setLinkDate(new Date());
            deviceChannelInfo.setUserId(msgAgreement.getUserId());
            this.addClientCache(msgAgreement.getUserId(),socketChannel, deviceChannelInfo);
        }

    }


    /**
     * 【 客户端连接服务端 】
     *
     * @author zhoujun
     * @date 2021/11/26 15:30
     * @param ctx
     * @return void
     **/
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        SocketChannel channel = (SocketChannel) ctx.channel();
        this.channelGroup.add(channel);
        logger.info("客户端上线,ip:{},channelId:{}",channel.remoteAddress(),channel.id());
    }


    /**
     * 【 客户端断开连接 】
     *
     * @author zhoujun
     * @date 2021/11/26 15:29
     * @param ctx
     * @return void
     **/
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        logger.info("客户端下线,ip:{},channelId:{}",channel.remoteAddress(), ctx.channel().id());
        this.channelGroup.remove(channel);
        removeClientCache(ctx.channel());
    }

    /**
     * 【 检查客户端连接状态(心跳检测) 】
     *
     * @author zhoujun
     * @date 2021/11/26 14:49
     * @param ctx
 * @param evt
     * @return void
     **/
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if(evt instanceof IdleStateEvent){
            IdleStateEvent event = (IdleStateEvent)evt;
            String eventType = null;
            switch (event.state()){
                case READER_IDLE:
                    eventType="读空闲";
                    break;
                case WRITER_IDLE:
                    eventType="写空闲";
                    break;
                case ALL_IDLE:
                    eventType="读写空闲";
                    Channel channel = ctx.channel();
                    //关闭无用channel，避免浪费资源
                    channel.close();
                    this.removeClientCache(ctx.channel());
                    logger.info(ctx.channel().remoteAddress()+"-"+eventType);
                    break;
            }
            
        }

    }


    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        logger.info("handlerAdded:"+ctx.channel().id().asLongText());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        logger.info("handlerRemoved:"+ctx.channel().id().asLongText());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("连接发生异常",cause);
        ctx.close();
        this.removeClientCache(ctx.channel());
    }

    /**
     * 【 程序关闭时清除客户端信息缓存 】
     *
     * @author zhoujun
     * @date 2021/11/26 15:29
     * @param
     * @return void
     **/
    @Override
    public void destroy() throws Exception {
        logger.info("服务器netty关闭");
        channelGroup.forEach(item->{
            this.removeClientCache(item);
        });
    }

    /**
     * 【 添加客户端信息缓存 】
     *
     * @author zhoujun
     * @date 2021/11/26 15:47
     * @param channel
     * @param deviceChannelInfo
     * @return void
     **/
    private void addClientCache(String userId,SocketChannel channel, DeviceChannelInfo deviceChannelInfo) {
        //设置二级key过期时间为10分钟后
        deviceChannelInfo.setExpiredTime(LocalDateTime.now().plusMinutes(10).toDate());
        this.cacheService.getRedisUtil().pushObj(deviceChannelInfo);
        CacheUtil.cacheChannel.put(userId, channel);
    }

    /**
     * 【 清除客户端信息缓存 】
     *
     * @author zhoujun
     * @date 2021/11/26 15:28
     * @param channel
     * @return void
     **/
    private void removeClientCache(Channel channel) {
        this.cacheService.getRedisUtil().remove(channel.id().toString());
        CacheUtil.cacheChannel.remove(channel.id().toString(), channel);
    }

    /**
     * 【 发送信息给客户端 】
     *
     * @author zhoujun
     * @date 2021/11/26 15:53
     * @param channel
     * @param message
     * @return void
     **/
    private void sendToClient(Channel channel, String message) {
        channel.writeAndFlush(new TextWebSocketFrame(message));
    }

    /**
     * 【 定时删除已过期的设备信息 】
     *
     * @author zhoujun
     * @date 2021/11/26 17:12
     * @param
     * @return void
     **/
    //每小时0分、30分执行
    //@Scheduled(cron = "0 0,30 * * * ? *")
    //每5分钟执行
    //@Scheduled(fixedRate=1000*60*5)
    private void timingDeleteClientCache(){
        logger.info("定时执行删除过期的客户端信息开始...");
        List<DeviceChannelInfo> deviceChannelInfos = cacheService.getRedisUtil().popList();
        AtomicInteger delCount = new AtomicInteger();
        deviceChannelInfos.forEach(item->{
            if (item.getExpiredTime().after(new Date())){
                return;
            }
            this.cacheService.getRedisUtil().remove(item.getChannelId());
            CacheUtil.cacheChannel.remove(item.getChannelId());
            delCount.addAndGet(1);
        });
        logger.info("定时执行删除过期的客户端信息结束...过期数据条数:{}",delCount.get());

    }
}

